package com.grf.hadoop.map_join;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;

public class DistributeCacheMapper extends Mapper<LongWritable, Text, Text, NullWritable> {

    private HashMap<String, String> pdMap = new HashMap<>();
    private Text k = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        //进行join操作
        //读取一行
        String line = value.toString();
        //切割
        String[] fields = line.split(" ");
        //获取ID
        String pid = fields[1];
        //取出pname
        String pname = pdMap.get(pid);
        //拼接
        line = line + " " + pname;
        k.set(line);
        //写出
        context.write(k, NullWritable.get());

    }

    /**
     * 缓存小表
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        //缓存小表
        URI[] cacheFiles = context.getCacheFiles();
        String path = cacheFiles[0].getPath().toString();
        BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream("")
                ,"UTF-8"));
        String line;
        while (StringUtils.isNotEmpty(line = reader.readLine())) {
            //切割
            String[] field = line.split(" ");
            pdMap.put(field[0],field[1]);
        }
        //关闭资源
        IOUtils.closeStream(reader);
    }
}
